package com.lld.im.tcp.reciver;

import com.alibaba.fastjson.JSONObject;
import com.lld.im.codec.proto.MessagePack;
import com.lld.im.common.constant.Constants;
import com.lld.im.tcp.reciver.process.BaseProcess;
import com.lld.im.tcp.reciver.process.ProcessFactory;
import com.lld.im.tcp.utils.MqFactory;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.io.IOException;

/**
 * @description:接收消息并进行处理。让我解释一下代码的功能：
 * 首先，通过 MqFactory.getChannel() 方法获取一个 Channel 对象，用于与 RabbitMQ 建立连接并进行通信。
 * 然后，通过 channel.queueDeclare() 方法声明一个队列，该队列用于接收消息。这里使用了常量 Constants.RabbitConstants.MessageService2Im + brokerId 作为队列的名称。
 * 接下来，通过 channel.queueBind() 方法将队列绑定到交换机。这里使用了常量 Constants.RabbitConstants.MessageService2Im 作为交换机的名称，brokerId 作为绑定键。
 * 使用 channel.basicConsume() 方法开始消费消息。该方法接受队列名称、是否自动确认消息和一个 DefaultConsumer 对象作为参数。
 * 在 DefaultConsumer 的 handleDelivery() 方法中处理接收到的消息。首先将消息的字节数组转换为字符串，并进行日志记录。然后，将消息字符串解析为 MessagePack 对象，并根据消息的命令获取相应的消息处理器（messageProcess）。最后，调用 messageProcess.process() 方法处理消息。
 * 如果消息处理过程中出现异常，将打印异常信息，并使用 channel.basicNack() 方法对消息进行拒绝（nack）操作，将消息重新放回队列中等待处理。
 * 如果消息处理成功，使用 channel.basicAck() 方法对消息进行确认（ack）操作，告诉 RabbitMQ 可以删除该消息。
 * @author: lld
 * @version: 1.0
 */
@Slf4j
public class MessageReciver {

    private static String brokerId;

    private static void startReciverMessage() {
        try {
            Channel channel = MqFactory
                    .getChannel(Constants.RabbitConstants.MessageService2Im + brokerId);
            channel.queueDeclare(Constants.RabbitConstants.MessageService2Im + brokerId,
                    true, false, false, null
            );
            channel.queueBind(Constants.RabbitConstants.MessageService2Im + brokerId,
                    Constants.RabbitConstants.MessageService2Im, brokerId);

            channel.basicConsume(Constants.RabbitConstants
                            .MessageService2Im + brokerId, false,
                    new DefaultConsumer(channel) {
                        @Override
                        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                            try {
                                String msgStr = new String(body);
                                log.info(msgStr);
                                MessagePack messagePack =
                                        JSONObject.parseObject(msgStr, MessagePack.class);
                                BaseProcess messageProcess = ProcessFactory
                                        .getMessageProcess(messagePack.getCommand());
                                messageProcess.process(messagePack);

                                channel.basicAck(envelope.getDeliveryTag(),false);

                            }catch (Exception e){
                                e.printStackTrace();
                                channel.basicNack(envelope.getDeliveryTag(),false,false);
                            }
                        }
                    }
            );
        } catch (Exception e) {

        }
    }

    public static void init() {
        startReciverMessage();
    }

    public static void init(String brokerId) {
        if (StringUtils.isBlank(MessageReciver.brokerId)) {
            MessageReciver.brokerId = brokerId;
        }
        startReciverMessage();
    }


}
